package com.minelx.datamicroscope.core.index;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.minelx.datamicroscope.common.Configurations;
import com.minelx.datamicroscope.common.EnrichedIterator;
import com.minelx.datamicroscope.core.condition.Condition;
import com.minelx.datamicroscope.core.condition.RawCondition;
import com.minelx.datamicroscope.core.query.AliasedKey;
import com.minelx.datamicroscope.core.query.CollectingResource;
import com.minelx.datamicroscope.core.query.TrickleQuery;
import com.minelx.datamicroscope.core.values.CommonField;
import com.minelx.hellopandas.core.DataFrameDescription;
import com.minelx.hellopandas.core.Pandas;
import com.minelx.hellopandas.core.PandasMethodProvider;
import com.minelx.statementsplus.cha.base.TaskCommitment;
import io.gitee.minelx.commontools.factor.CloseableIterator;
import io.gitee.minelx.filesystem.Directory;
import io.gitee.minelx.filesystem.File;
import com.minelx.statementsplus.cha.base.ChaSession;
import io.gitee.minelx.commontools.factor.Factor;
import io.gitee.minelx.commontools.factor.RegisterIterator;
import io.gitee.minelx.filesystem.TempPath;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.Map;
import java.util.Set;

import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.emptyList;
import static java.util.stream.Collectors.*;

class OnlineIndex implements Index {

    public static final int DEFAULT_WORKER_COUNTS = 5;

    public static final JSONArray indexDescriptions = Configurations.of("index_descriptions").toJSON();

    public static final Logger logger = LoggerFactory.getLogger(Index.class);

    private final PandasMethodProvider pandasMethodProvider;

    private final Directory cacheFilesDirectory;
    private final CollectingResource collectingResource;
    private final File destination;

    private final ChaSession session;

    private final List<RawCondition> conditions;

    private final List<AliasedKey> remainingKeys;

    OnlineIndex(PandasMethodProvider pandasMethodProvider,
                Directory cacheFilesDirectory,
                ChaSession session,

                List<AliasedKey> remainingKeys,
                List<RawCondition> conditions,
                CollectingResource collectingResource,
                File destination) {
        this.pandasMethodProvider = pandasMethodProvider;
        this.session = session;
        this.remainingKeys = remainingKeys;
        this.conditions = conditions;
        this.cacheFilesDirectory = cacheFilesDirectory;
        this.collectingResource = collectingResource;
        this.destination = destination;
    }

    // FIXME wait for:  use Spliterator instead of EnrichedIterator
    @Override
    public void list() {
        // create factors from raw conditions
        List<Factor<Condition>> factors = conditions.stream().map(RawCondition::toFactor).collect(toList());
        // convert from factors to RegisterIterator
        try (CloseableIterator<List<Condition>> registerIterator = RegisterIterator.create(factors)) {
            // verify if there is empty factor contains then write out the destination as a title with empty content.
            if (!registerIterator.hasNext()) {
                writeTitleToDestination(this.collectingResource, this.destination, remainingKeys);
                return;
            }
            // convert from each flatten condition to a trickle query
            EnrichedIterator<TrickleQuery> queries = EnrichedIterator.fromUnSafeIterator(registerIterator)
                    .map(conditions -> this.collectingResource.appendAndBuild(remainingKeys, conditions));
            // fire queries
            doTrickleQuery(pandasMethodProvider,
                    cacheFilesDirectory,
                    session,
                    queries,
                    this.destination,
                    remainingKeys.stream().map(AliasedKey::getKey).collect(toSet()),
                    representTitle(this.collectingResource, remainingKeys));
        }
    }

    private static void writeTitleToDestination(CollectingResource collectingResource, File destination, List<AliasedKey> remainingKeys) {
        String content = representTitle(collectingResource, remainingKeys);
        try {
            destination.write(content, StandardCharsets.UTF_8);
        } catch (IOException e) {
            throw new RuntimeException("error while writing csv content.", e);
        }
    }

    private static String representTitle(CollectingResource collectingResource, List<AliasedKey> remainingKeys) {
        TrickleQuery trickleQuery = collectingResource.appendAndBuild(remainingKeys, emptyList());
        return trickleQuery.representTitleExceptPrimary();
    }

    public static void doTrickleQuery(PandasMethodProvider pandasMethodProvider,
                                      Directory cacheFilesDirectory,
                                      ChaSession session,

                                      EnrichedIterator<TrickleQuery> trickleQueries,
                                      File destination,
                                      Set<CommonField> connectedKeys,
                                      String representTitle) {
        int nThreads = workerCounts(connectedKeys);
        logger.info("start working on multi-thread, fields connected on: " +
                connectedKeys.stream().map(CommonField::path).collect(joining(", ", "(", ")")) +
                ", nThread counts: " + nThreads);

        try (OutputStream outputStream = destination.outputStream(StandardOpenOption.APPEND);
             BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(outputStream, UTF_8))) {
            // write out title content
            writer.write(representTitle + "\n");

            trickleQueries.consumeAll(
                    trickleQuery -> queryAndSave(pandasMethodProvider, cacheFilesDirectory, session, writer, trickleQuery),
                    nThreads);
        } catch (IOException e) {
            throw new RuntimeException("error while doing...", e);
        }
    }

    public static void queryAndSave(PandasMethodProvider pandasMethodProvider,
                                    Directory cacheFilesDirectory,
                                    ChaSession session,

                                    BufferedWriter destinationAppendingWriter,
                                    TrickleQuery trickleQuery) {
        try {
            String content = queryToContent(pandasMethodProvider, cacheFilesDirectory, session, trickleQuery);
            destinationAppendingWriter.write(content);
        } catch (IOException e) {
            throw new RuntimeException("error while writing content to destination appending writer.", e);
        }
    }

    public static String queryToContent(PandasMethodProvider pandasMethodProvider,
                                        Directory cacheFilesDirectory,
                                        ChaSession session,

                                        TrickleQuery trickleQuery) {
        try (TempPath temp = TempPath.random(cacheFilesDirectory)) {
            TaskCommitment taskCommitment = trickleQuery.explainForCommitment(session);
            session.download(taskCommitment, temp.file());

            // append nothing if this trickle query point to nothing
            if (temp.file().empty()) {
                return "";
            }

            // ordered by specified columns
            Map<String, String> dtypes = trickleQuery.dtypesOfQueriedContent();
            Pandas.of(pandasMethodProvider).dataFrame(DataFrameDescription.create(temp.file().path().uri(), dtypes))
                    .retains(trickleQuery.representColumnsExceptPrimary0());
            // title removed
            String titleRemoved = csvContentOf(lineSeparatorGeneralized(temp.file().read(UTF_8)));
            // write out as a line
            return lineAppendedBehind(titleRemoved);
        } catch (IOException e) {
            throw new RuntimeException("error while reading downloaded file content.", e);
        }
    }

    public static String lineSeparatorGeneralized(String text) {
        return text.replace("\r\n", "\n");
    }

    public static String csvContentOf(String retainedCsvText) {
        return retainedCsvText.substring(retainedCsvText.indexOf('\n') + 1);
    }

    public static String lineAppendedBehind(String text) {
        if (text.endsWith("\n")) {
            return text;
        }
        return text + "\n";
    }

    public static int workerCounts(Set<CommonField> commonFields) {
        JSONObject config = indexDescriptions.stream()
                .map(item -> ((JSONObject) item))
                .filter(jsonObject -> {
                    List<String> keys = jsonObject.getJSONArray("keys").toJavaList(String.class);
                    Set<CommonField> indexFields = keys.stream()
                            .map(CommonField::from)
                            .collect(toSet());
                    return indexFields.equals(commonFields);
                })
                .findFirst().orElse(getDefaultConfig(commonFields));
        return config.getIntValue("workerCounts");
    }

    public static JSONObject getDefaultConfig(Set<CommonField> keys) {
        JSONObject result = new JSONObject();
        result.put("keys", keys.stream()
                .map(CommonField::path)
                .collect(toList()));
        result.put("workerCounts", DEFAULT_WORKER_COUNTS);
        return result;
    }
}
